package com.xuecheng.media.domain.jobhandler;

import com.xuecheng.media.domain.model.entity.MediaFiles;
import com.xuecheng.media.domain.model.entity.MediaProcess;
import com.xuecheng.media.domain.model.entity.UploadRecord;
import com.xuecheng.media.domain.service.MediaFilesService;
import com.xuecheng.media.domain.service.MediaProcessService;
import com.xuecheng.media.domain.service.UploadRecordService;
import com.xuecheng.media.infrastructure.enums.MimeTypeEnum;
import com.xuecheng.media.infrastructure.enums.UploadStatusEnum;
import com.xuecheng.media.infrastructure.enums.VideTypeEnum;
import com.xuecheng.media.infrastructure.util.MinioUtil;
import com.xuecheng.media.infrastructure.util.UploadUtil;
import com.xuecheng.media.infrastructure.util.VideoUtil;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @Author: LiJunXi
 * @Description:
 * @Date: 2024-08-17  10:35
 */
@Slf4j
@Component
public class ShardingJob {

    @Autowired
    private MediaProcessService mediaProcessService;

    @Autowired
    private MediaFilesService mediaFilesService;

    @Autowired
    private UploadRecordService uploadRecordService;

    @Autowired
    private MinioUtil minioUtil;

    // 存储视频
    @Value("${minio.bucket.videofiles}")
    private String bucketVideo;

    /**
     * 处理视频转码
     * @throws InterruptedException
     */
    @XxlJob("videoJobHandler")
    public void videoJobHandler() throws InterruptedException {
        Integer shardIndex = XxlJobHelper.getShardIndex();
        Integer shardTotal = XxlJobHelper.getShardTotal();
        log.info("=================================================================");
        log.info("执行视频转码任务调度 ==> index:{}, total:{}", shardIndex, shardTotal);

        // 取出cpu核心数作为一次处理数据的条数
        Integer processors = Runtime.getRuntime().availableProcessors();
        // 获取当前分片能够处理的视频列表
        List<MediaProcess> waitHandleTaskList = mediaProcessService.getWaitHandleTaskList(shardIndex, shardTotal, processors);
        publicExecute(waitHandleTaskList);

    }

    private void publicExecute(List<MediaProcess> waitHandleTaskList) throws InterruptedException {
        if(waitHandleTaskList.isEmpty()){
            return;
        }
        // 创建最多处理数量个单位的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(waitHandleTaskList.size());
        // 计数器
        CountDownLatch countDownLatch = new CountDownLatch(waitHandleTaskList.size());
        log.info("线程池开始处理待处理视频列表");
        for (MediaProcess mediaProcess : waitHandleTaskList) {
            executorService.submit(() -> {
                try {
                    handleTask(mediaProcess);
                } catch (Exception e){
                    log.error("执行转码任务失败，原因：{}", e.getMessage());
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        // 30s后没有完成就结束任务，防止无限等待
        countDownLatch.await(30, TimeUnit.MINUTES);
    }

    private void handleTask(MediaProcess mediaProcess){
        log.info("开始处理视频：{}", mediaProcess);
        Long taskId = mediaProcess.getId();
        // 获取锁防止多实例重复消费（乐观锁）
        Boolean flag = mediaProcessService.startTask(taskId);
        if(!flag){
            return;
        }

        File fileObj = minioUtil.getObj(mediaProcess.getBucket(), mediaProcess.getFilePath());
        if (Objects.isNull(fileObj)){
            // 需要转码的视频文件不存在，更新状态
            log.error("转码视频文件不存在");
            mediaProcessService.updateProcess(mediaProcess.buildError("转码视频文件不存在"));
            return;
        }
//        // 临时文件
//        File tmpFile = null;
//        try {
//            tmpFile = File.createTempFile("mp4", ".mp4");
//
//
//        } catch (Exception e){
//            log.error("创建临时文件失败");
//            mediaProcessService.updateProcess(mediaProcess.buildError("创建临时文件失败"));
//            return;
//        }

        String filename = mediaProcess.getFilename().split("\\.")[0];
        VideoUtil videoUtil = new VideoUtil(fileObj.getAbsolutePath(), filename, VideTypeEnum.MP4);
        String res = videoUtil.generate();
        if (!res.equals("success")){
            // 转码失败
            log.error("视频转码失败");
            mediaProcessService.updateProcess(mediaProcess.buildError("视频转码失败"));
            return;
        }

        // 保存到minio
        String filePath = UploadUtil.getDefaultAbsolutePath() + filename + "." + VideTypeEnum.MP4.getType();
        mediaFilesService.addMediaFileToMinio(fileObj.getAbsolutePath(), MimeTypeEnum.MP4.getType(), mediaProcess.getBucket(),
                filePath);
        // 标记视频已经处理了
        mediaProcessService.updateProcess(mediaProcess.buildSuccess());
        // 更新媒资信息
        MediaFiles mediaFiles = new MediaFiles();
        mediaFiles.setId(mediaProcess.getFileId());
        mediaFiles.setFilename(filename + "." + VideTypeEnum.MP4.getType());
        mediaFiles.setFilePath(filePath);
        mediaFilesService.updateMediaFiles(mediaFiles);

        log.info("视频转码任务调度完成");
    }

    /**
     * 处理视频补偿机制
     * 如果有线程抢占了某个视频的处理任务，如果线程处理过程中挂掉了，该视频的状态将会一直是处理中，其它线程将无法处理，这个问题需要用补偿机制
     */
    @XxlJob("videoCompensate")
    public void videoCompensate() throws InterruptedException {
        Integer shardIndex = XxlJobHelper.getShardIndex();
        Integer shardTotal = XxlJobHelper.getShardTotal();
        log.info("=================================================================");
        log.info("执行视频补偿任务调度 ==> index:{}, total:{}", shardIndex, shardTotal);
        // 取出cpu核心数作为一次处理数据的条数
        Integer processors = Runtime.getRuntime().availableProcessors();
        List<MediaProcess> compensateTaskList = mediaProcessService.getCompensateTaskList(shardIndex, shardTotal, processors);
        publicExecute(compensateTaskList);
        log.info("视频补偿任务调度完成");

    }


    /**
     * 文件分片处理
     * 上传一个文件进行分块上传，上传一半不传了，之前上传到minio的分块文件要清理
     * @throws InterruptedException
     */
    @XxlJob("videoChunkClearHandler")
    public void videoChunkClearHandler() throws InterruptedException {
        Integer shardIndex = XxlJobHelper.getShardIndex();
        Integer shardTotal = XxlJobHelper.getShardTotal();
        log.info("=================================================================");
        log.info("执行分片文件垃圾清理任务调度 ==> index:{}, total:{}", shardIndex, shardTotal);
        // 取出cpu核心数作为一次处理数据的条数
        Integer processors = Runtime.getRuntime().availableProcessors();
        // 获取需要清除的分片文件列表
        List<UploadRecord> waitClearRecordList = uploadRecordService.getWaitClearRecordList(shardIndex, shardTotal, processors);
        if (waitClearRecordList.isEmpty()){
            return;
        }
        try {
            // 创建最多处理数量个单位的线程池
            ExecutorService executorService = Executors.newFixedThreadPool(waitClearRecordList.size());
            // 计数器
            CountDownLatch countDownLatch = new CountDownLatch(waitClearRecordList.size());
            for (UploadRecord uploadRecord : waitClearRecordList) {
                executorService.execute(() -> {
                    try {
                        String fileMd5 = uploadRecord.getFileId();
                        String chunkPath = UploadUtil.getChunkPath(fileMd5);
                        Boolean res = minioUtil.removeFile(bucketVideo, chunkPath);
                        if (res){
                            // 更新上传文件记录状态
                            uploadRecord.setStatus(UploadStatusEnum.OVER.getType());
                            uploadRecordService.updateById(uploadRecord);
                        }
                    } catch (Exception e){
                        log.error("处理视频分片任务调度失败，原因：{}", e.getMessage());
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            //等待,给一个充裕的超时时间,防止无限等待，到达超时时间还没有处理完成则结束任务
            countDownLatch.await(30, TimeUnit.SECONDS);
        } catch (Exception e){
            log.error("处理视频分片任务调度失败，原因：{}", e.getMessage());
        }
        log.info("视频分片任务调度完成");

    }


}
